package site.teamo.biu.flink.function.process;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.codehaus.commons.compiler.CompileException;
import org.codehaus.commons.compiler.IScriptEvaluator;
import org.codehaus.janino.ScriptEvaluator;
import site.teamo.biu.flink.sdk.annotation.function.BFProcessFunction;
import site.teamo.biu.flink.sdk.function.AbstractProcessFunction;
import site.teamo.biu.flink.sdk.job.JobContext;

/**
 * @author haocongshun
 * @date 2023/07/27 14:20:47
 */
@BFProcessFunction
public class ScriptFunction extends AbstractProcessFunction<Object, Object> {

    private ScriptExecutor<Object, Object> executor;

    public ScriptFunction(String name, JobContext context) {
        super(name, context);

    }

    @Override
    public void open(Configuration parameters) throws Exception {
        String code = this.context.getConfig(this.name).getString("script");
        String type = this.context.getConfig(this.name).getString("type");
        switch (type) {
            case "JAVA":
                this.executor = new JavaExecutor<Object, Object>(code);
                break;
            default:
                throw new RuntimeException("不支持的代码脚本");
        }
    }

    @Override
    public void processElement(Object in, ProcessFunction<Object, Object>.Context context, Collector<Object> collector) throws Exception {
        if (executor == null) {
            return;
        }
        collector.collect(executor.execute(in));
    }


    private static abstract class ScriptExecutor<IN, OUT> {
        protected final String code;

        public ScriptExecutor(String code) {
            this.code = code;
        }

        public abstract OUT execute(IN in) throws Exception;
    }

    private static class JavaExecutor<IN, OUT> extends ScriptExecutor<IN, OUT> {

        private IScriptEvaluator evaluator;

        public JavaExecutor(String code) throws CompileException {
            super(code);
            this.evaluator = new ScriptEvaluator();
            this.evaluator.setReturnType(Object.class);
            this.evaluator.setParameters(new String[]{"record"}, new Class[]{Object.class});
            this.evaluator.cook(code);
        }

        @Override
        @SuppressWarnings("unchecked")
        public OUT execute(IN in) throws Exception {
            return (OUT) evaluator.evaluate(new Object[]{in});
        }
    }
}
